package com.imooc.bilibili.config;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.MQConstant;
import com.imooc.bilibili.domain.Danmu;
import com.imooc.bilibili.domain.UserFollowing;
import com.imooc.bilibili.domain.UserMoment;
import com.imooc.bilibili.service.DanmuService;
import com.imooc.bilibili.service.UserFollowingService;
import com.imooc.bilibili.websocket.WebSocketService;
import com.mysql.cj.util.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.List;

/**
 * RocketMQ 配置类
 *
 * @author xiexu
 * @create 2022-06-15 18:41
 */
@Configuration
public class RocketMQConfig {

    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowingService userFollowingService;

    @Autowired
    private DanmuService danmuService;

    /**
     * 用户动态相关的生产者
     *
     * @return
     * @throws Exception
     */
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws Exception {
        // 新建一个 MQ 分组
        DefaultMQProducer producer = new DefaultMQProducer(MQConstant.GROUP_MOMENTS);
        // 设置 nameserver 的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动生产者
        producer.start();
        return producer;
    }

    /**
     * 用户动态相关的消费者
     *
     * @return
     * @throws Exception
     */
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQConstant.GROUP_MOMENTS);
        // 设置 nameserver 的地址
        consumer.setNamesrvAddr(nameServerAddr);
        // 消费者需要订阅的是哪个生产者的topic
        consumer.subscribe(MQConstant.TOPIC_MOMENTS, "*");
        // 消费者监听消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             *
             * @param msgs 消息
             * @param context 消息上下文
             * @return
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 每次发布动态都是向 MQ 发送一条消息，所以这里只取第一个元素就行
                MessageExt msg = msgs.get(0);
                if (msg == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 获取msg里面的实体数据
                String bodyStr = new String(msg.getBody());
                // 将接收到的消息转换成UserMoment实体类
                UserMoment userMoment = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoment.class);
                // 获取动态的用户id
                Long userId = userMoment.getUserId();
                // 当前用户是userId，通过查询关注他的粉丝，从而给这些粉丝推送消息
                List<UserFollowing> fanList = userFollowingService.getUserFans(userId);
                for (UserFollowing fan : fanList) {
                    // 给每位粉丝发送新的动态消息
                    String key = "subscribed-" + fan.getUserId();
                    // 获取的是该粉丝所订阅的所有up主的动态消息列表，但是是字符串格式，需要进行格式转换
                    String subscribedListStr = redisTemplate.opsForValue().get(key);
                    // 动态消息列表
                    List<UserMoment> subscribedList;
                    // 如果该粉丝的动态消息列表为null，那就创建一个新的列表
                    if (StringUtils.isNullOrEmpty(subscribedListStr)) {
                        subscribedList = new ArrayList<>();
                    } else { // 如果该粉丝的动态消息列表不为null
                        // 将字符串转换成列表
                        subscribedList = JSONArray.parseArray(subscribedListStr, UserMoment.class);
                    }
                    // 将我们新生成的动态消息放入消息列表中
                    subscribedList.add(userMoment);
                    // 重新保存到redis里面
                    redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));
                }
                // 返回消息消费成功的状态
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        return consumer;
    }

    /**
     * 弹幕相关的生产者
     *
     * @return
     * @throws Exception
     */
    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer(MQConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动Producer实例
        producer.start();
        return producer;
    }

    /**
     * 弹幕相关的消费者
     *
     * @return
     * @throws Exception
     */
    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        consumer.setNamesrvAddr(nameServerAddr);
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe(MQConstant.TOPIC_DANMUS, "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                if (msg == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(msg.getBody());
                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");
                // 根据sessionId获取对应的webSocketService服务
                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);

                // 判断会话是否还处于打开状态
                if (webSocketService.getSession().isOpen()) {
                    try {
                        // 服务器发送消息给客户端
                        webSocketService.sendMessage(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

    /**
     * 异步保存弹幕生产者
     *
     * @return
     * @throws Exception
     */
    @Bean("asyncadddanmusProducer")
    public DefaultMQProducer asyncAddDanmusProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer(MQConstant.GROUP_ASYNCADDDANMUS);
        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动Producer实例
        producer.start();
        return producer;
    }

    /**
     * 异步保存弹幕消费者
     *
     * @return
     * @throws Exception
     */
    @Bean("asyncadddanmusConsumer")
    public DefaultMQPushConsumer asyncAddDanmusConsumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQConstant.GROUP_ASYNCADDDANMUS);
        // 设置NameServer的地址
        consumer.setNamesrvAddr(nameServerAddr);
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe(MQConstant.TOPIC_ASYNCADDDANMUS, "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                if (msg == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(msg.getBody());
                // 将接收到的消息转换成DanMu实体类
                Danmu danmu = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), Danmu.class);
                // 异步保存弹幕
                danmuService.asyncAddDanmu(danmu);

                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

}
